#ifndef __M__CHANNEL_H_
#define __M__CHANNEL_H_
#include "consumer.hpp"

#include "../mqCommon/proto/mq_message.pb.h"
#include "../mqCommon/proto/mq_proto.pb.h"
#include "../third/muduo/include/muduo/net/TcpConnection.h"
#include "../third/muduo/proto/codec.h"

#include <thread>
#include <mutex>
#include <condition_variable>

using namespace mq_message;
using namespace mq;
using namespace std;
using namespace muduo::net;

namespace mq
{
    using ProtobufCodecPtr = shared_ptr<ProtobufCodec>;               // 协议处理器
    using basicCommonResponsePtr = shared_ptr<basicCommonResponse>;   // 通用响应
    using basicConsumeResponsePtr = shared_ptr<basicConsumeResponse>; // 消息推送
    using Map = google::protobuf::Map<string, string>;

    // 客户端信道类
    class Channel
    {
    public:
        using ptr = shared_ptr<Channel>;
        Channel(const TcpConnectionPtr &conn,
                const ProtobufCodecPtr &codec)
            : _cid(UUIDHelper::uuid()), _conn(conn), _codec(codec) {}
        ~Channel()
        {
            // 释放信道开辟的资源
            if (_consumer.get())
                basicCancel(); // 取消订阅
        }
        // 0.打开/关闭信道
        bool openChannel()
        {
            // 1.构造请求
            openChannelRequest req;
            string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            // 2.发送请求给服务端，创建对应的信道
            // cout<<"a"<<endl;
             if(_conn.get()==nullptr){
                cout<<"_conn=nullptr"<<endl;
                return false;
            }
            if(!_conn->connected()){
                cout<<"unconnected"<<endl;
                return false;
            }
            _codec->send(_conn, req);
            bool ret=waitResponse(rid);
            return ret;
        }
        void closeChannel()
        {
            // 1.构造请求
            closeChannelRequest req;
            string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            // 2.发送请求给服务端，关闭对应的信道
            _codec->send(_conn, req);
            waitResponse(rid);
        }
        // 1.交换机的声明/删除
        bool declareExchange(const string &name, ExchangeType type, bool durable, bool auto_delete, Map &args)
        {
            // 1.构造请求
            declareExchangeRequest req;
            // a.哪个信道的哪个请求
            const string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            // b.交换机信息
            req.set_exchange_name(name);
            req.set_exchange_type(type);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            // 2.发送请求
            _codec->send(_conn, req); // 非阻塞接口，如果直接返回，可能交换机实际还没声明成功
            // 3.阻塞，直到收到基础响应（ok/不ok）
            // 4.返回ok/不ok
            return waitResponse(rid);
        }
        void deleteExchange(const string &name)
        {
            // 1.构造请求
            deleteExchangeRequest req;
            const string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(name);
            // 2.发送请求
            _codec->send(_conn, req);
            waitResponse(rid);
            return;
        }
        // 2.队列的声明/删除
        bool declareMsgQueue(const string &name, bool exclusive, bool durable, bool auto_delete, Map &args)
        {
            // 1.构造请求
            declareQueueRequest req;
            const string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(name);
            req.set_exclusive(exclusive);
            req.set_durable(durable);
            req.set_auto_delete(auto_delete);
            req.mutable_args()->swap(args);
            // 2.发送请求
            _codec->send(_conn, req); // 非阻塞接口，如果直接返回，可能队列实际还没声明成功
            // 3.阻塞，直到收到基础响应（ok/不ok）
            // 4.返回ok/不ok
            return waitResponse(rid);
        }
        void deleteMsgQueue(const string &name)
        {
            // 1.构造请求
            deleteQueueRequest req;
            const string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(name);
            // 2.发送请求
            _codec->send(_conn, req);
            waitResponse(rid);
            return;
        }
        // 3.绑定/解绑
        bool Bind(const string &ename, const string &qname, const string &binding_key)
        {
            // 1.构造请求
            queueBindRequest req;
            // a.哪个信道的哪个请求
            const string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            // b.绑定信息
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            req.set_binding_key(binding_key);
            // 2.发送请求
            _codec->send(_conn, req); // 非阻塞接口，如果直接返回，可能实际还没绑定成功
            // 3.阻塞，直到收到基础响应（ok/不ok）
            // 4.返回ok/不ok
            return waitResponse(rid);
        }
        void unBind(const string &ename, const string &qname)
        {
            // 1.构造请求
            queueUnBindRequest req;
            const string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            req.set_queue_name(qname);
            // 2.发送请求
            _codec->send(_conn, req);
            waitResponse(rid);
            return;
        }
        // 4.订阅队列/取消订阅
        bool basicConsume(const string &consumer_tag, const string &qname, bool auto_ack, const ConsumerCallback &cb)
        {
            // 1.检查是否重复订阅
            if (_consumer.get() != nullptr) // 当前信道已经订阅过队列了
            {
                DLOG("当前信道已订阅其它队列消息,无法重复订阅,consumer_tag[%s] | queue_name[%s]", _consumer->_tag.c_str(), _consumer->_qname.c_str());
                return false;
            }

            // 2.构造请求
            basicConsumeRequest req;
            const string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_consumer_tag(consumer_tag);
            req.set_queue_name(qname);
            req.set_auto_ack(auto_ack);

            // 3.发送请求
            _codec->send(_conn, req);
            if (!waitResponse(rid))
            {
                DLOG("订阅失败,该队列可能未声明,consumer_tag[%s] | queue_name[%s]", consumer_tag.c_str(), qname.c_str());
                return false;
            }
            else
                _consumer = make_shared<Consumer>(consumer_tag, qname, auto_ack, cb); // 4.订阅成功后，才创建订阅对象并管理

            return true;
        }
        void basicCancel()
        {
            // 1.检查是否有必要发送请求
            if (_consumer.get() == nullptr) // 还未订阅or已经取消过订阅了
                return;

            // 2.构造请求
            basicCancelRequest req;
            const string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_consumer_tag(_consumer->_tag);
            req.set_queue_name(_consumer->_qname);

            // 3.发送请求
            _codec->send(_conn, req);
            if (!waitResponse(rid))
            {
                DLOG("取消订阅失败,consumer_tag[%s] | queue_name[%s]", _consumer->_tag.c_str(), _consumer->_qname.c_str());
            }
            else
                _consumer.reset(); // 4.取消订阅成功后，将consumer重置，可以再次订阅

            return;
        }
        // 5.发布消息/确认消息
        void basicPublish(const string &ename, const BasicProperties *bp, const string &body)
        {
            // 1.构造请求
            basicPublishRequest req;
            const string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_exchange_name(ename);
            if (bp != nullptr)
            {
                req.mutable_properties()->set_id(bp->id());
                req.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                req.mutable_properties()->set_routing_key(bp->routing_key());
            }
            req.set_body(body);
            // 2.发送请求
            if(_conn.get()==nullptr){
                cout<<"_conn=nullptr"<<endl;
                return;
            }
            if(!_conn->connected()){
                cout<<"unconnected"<<endl;
                return;
            }
            _codec->send(_conn, req);
            waitResponse(rid);
            return;
        }
        void basicAck(const string &msg_id)
        {
            if (_consumer.get() == nullptr)
            {
                DLOG("还未订阅");
                return;
            }
            // 1.构造请求
            basicAckRequest req;
            const string rid = UUIDHelper::uuid();
            req.set_rid(rid);
            req.set_cid(_cid);
            req.set_queue_name(_consumer->_qname);
            req.set_message_id(msg_id);
            // 2.发送请求
            _codec->send(_conn, req);
            waitResponse(rid);
            return;
        }

    public:
        void putBasicCommonResponse(const basicCommonResponsePtr &resp) // 收到基础响应之后，放入对应信道的响应列表，并唤醒
        {
            // 1.加锁
            unique_lock<mutex> lock(mtx);
            // 2.添加响应信息
            _basic_resp.insert(make_pair(resp->rid(), resp));
            // 3.唤醒
            _cv.notify_all();
        }
        void consume(const basicConsumeResponsePtr &resp) // 收到消息之后，找到对应信道的消费者，异步处理线程执行回调函数cb
        {
            // 1.判断是否订阅队列
            if (_consumer.get() == nullptr)
            {
                DLOG("当前信道未订阅队列，收到未知队列推送消息： channel_cid[%s]  resp_cid[%s]  resp_ctag[%s]", _cid.c_str(), resp->cid(), resp->consumer_tag());
                return;
            }
            // 2.判断收到的推送消息消费标识与当前信道订阅标识是否匹配
            if (resp->consumer_tag() != _consumer->_tag)
            {
                DLOG("收到不属于当前信道的消息： channel_cid[%s]  channel_ctag[%s],resp_cid[%s]  resp_ctag[%s]", _cid.c_str(), _consumer->_tag.c_str(), resp->cid(), resp->consumer_tag());
                return;
            }

            // 3.执行回调
            _consumer->_callback(resp->consumer_tag(), resp->mutable_properties(), resp->body());
        }
        string get_cid() // 获取信道id
        {
            return _cid;
        }

    private:
        bool waitResponse(const string &rid) // 有些接口，需要在发 出请求后，等到收到响应后才能返回
        {
            unique_lock<mutex> lock(mtx);
            _cv.wait(lock, [&rid, this]()
                     { return _basic_resp.find(rid) != _basic_resp.end(); });
            basicCommonResponsePtr common_resp = _basic_resp.find(rid)->second;
            _basic_resp.erase(rid);
            return common_resp->ok();
        }

    private:
        string _cid;             // 信道ID
        Consumer::ptr _consumer; // 客户端信道关联的订阅者:为订阅客户端时，该信道订阅队列时创建；否则为发布客户端，不创建

        muduo::net::TcpConnectionPtr _conn; // 信道使用的网络通信连接(服务端连接)
        ProtobufCodecPtr _codec;            // protobuf协议处理句柄

        unordered_map<string, basicCommonResponsePtr> _basic_resp; // 请求对应的响应信息哈希表

        mutex mtx;
        condition_variable _cv; // 互斥锁&条件变量：接口之间需要同步
    };

    // 信道管理类：信道的增删查
    class ChannelManager
    {
    public:
        using ptr = shared_ptr<ChannelManager>;
        ChannelManager() {}
        ~ChannelManager() {
        }
        Channel::ptr openChanel(const TcpConnectionPtr &conn,
                                const ProtobufCodecPtr &codec) // 打开信道,返回channel::ptr
        {
            unique_lock<mutex> lock(mtx);
            // 创建信道，添加信道管理
            Channel::ptr channel = make_shared<Channel>(conn, codec);
            string cid = channel->get_cid();
            _channels.insert(make_pair(cid, channel));
            return channel;
        }
        void closeChannel(const Channel::ptr &channel) // 关闭信道
        {
            unique_lock<mutex> lock(mtx);
            if(channel.get()==nullptr)
            {
                cout<<"channel.get()==nullptr"<<endl;
            }
            string cid=channel->get_cid();
            _channels.erase(cid);
        }
        Channel::ptr getChannel(const string &cid) // 获取信道
        {
            unique_lock<mutex> lock(mtx);
            auto it = _channels.find(cid);
            if (it == _channels.end())
            {
                return Channel::ptr();
            }
            return it->second;
        }

    private:
        mutex mtx;                                     // 加锁保护
        unordered_map<string, Channel::ptr> _channels; // 根据信道id找到对应信道
    };
}
#endif
